package com.demo;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.ArrayBlockingQueue;

public class BrokerServer implements Runnable {
    public static int SERVICE_PORT = 8089;

    private final Socket socket;

    private final static int QUEUE_SIZE = 10;

    private static ArrayBlockingQueue<String> messageQueue = new ArrayBlockingQueue<>(QUEUE_SIZE);

    public BrokerServer(Socket socket) {
        this.socket = socket;
    }

    @Override
    public void run() {
        try (
                BufferedReader in = new BufferedReader(new InputStreamReader(
                        socket.getInputStream()));
                PrintWriter out = new PrintWriter(socket.getOutputStream())
        ) {
            while (true) {
                String str = in.readLine();
                if (str == null) {
                    continue;
                }
                System.out.println("get msg:" + str);
                if (str.equals("CONSUME")) {
                    String message = consume();
                    out.println(message);
                    out.flush();
                } else {
                    produce(str);
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    // 生产消息
    private  void produce(String msg) {
        if (messageQueue.offer(msg)) {
            System.out.println("msg:" + msg + "count:" + messageQueue.size());
        } else {
            System.out.println("this is full");
        }
        System.out.println("=======================");
    }

    private   String consume() {
        String msg = messageQueue.poll();
        if (msg != null) {
            System.out.println("msg:" + msg + "count:" + messageQueue.size());
        } else {
            System.out.println("null");
        }
        System.out.println("=======================");

        return msg;
    }

    public static void main(String[] args) throws Exception {
        ServerSocket server = new ServerSocket(SERVICE_PORT);
        while (true) {
            BrokerServer brokerServer = new BrokerServer(server.accept());
            new Thread(brokerServer).start();
        }
    }
}
